Add basic asyncio support #43944
Add basic asyncio support #43944
Conversation
dstandish
requested review from
potiuk,
ashb and
jedcunningham
as code owners
|
Is this related to AIP-70? |
I think you prob know the answer. The AIP deals with asyncio, so yes strictly speaking related in that sense. But not "part of". That AIP has been in draft a long time not sure actively worked on. Anyway, I don't think what I'm doing here really requires an AIP. |
|
There is a similar draft PR open as POC for it though not active. So just wanted to confirm. Thanks for the details. |
|
@pierrejeambrun A FastAPI dep to give an async session would be good, but I think anything in the models etc should be passed an explicit session. |
|
NICE! Yeah. I really like to see it working with mysql compatibility and benchmarks 📦 (because of course MySQL is our beloved database) |
i would say let's wait and see how / where / when we need it |
dstandish
force-pushed
the
add-async-engine-and-session
branch
from
0af9d4d
to
620902e
Compare
|
Lovely! Thank you for adding this, will try to start using this soon :) |
dstandish
force-pushed
the
add-async-engine-and-session
branch
from
5765c6c
to
8b5ab77
Compare
This is meant to be sort of a hello world for asyncio support in airflow. It will be refined and extended in the future. But I think airflow ultimately really needs to go in this direction: in the new REST API, in the new AIP-72 internal API server, in triggers, and ultimately, in the scheduler.
dstandish
force-pushed
the
add-async-engine-and-session
branch
from
8b5ab77
to
6791e41
Compare
|
@dstandish in lowest direct dep tests, we may need to limit minimum version for More info on this slack thread. |
| session = create_async_session() | ||
| session.add(Log(event="hihi1234")) | ||
| await session.commit() | ||
| l = await session.scalar(select(Log).where(Log.event == "hihi1234").limit(1)) # noqa: E741 |
There was a problem hiding this comment.
Can avoid this noqa by just change the var name :)
| l = await session.scalar(select(Log).where(Log.event == "hihi1234").limit(1)) # noqa: E741 | ||
| assert l.event == "hihi1234" |
There was a problem hiding this comment.
| l = await session.scalar(select(Log).where(Log.event == "hihi1234").limit(1)) # noqa: E741 | |
| assert l.event == "hihi1234" | |
| log_result = await session.scalar(select(Log).where(Log.event == "hihi1234").limit(1)) # noqa: E741 | |
| assert log_result.event == "hihi1234" |
This is meant to be sort of a hello world for asyncio support in airflow. It will be refined and extended in the future. E.g. probably we would add more config flexibility re the libraries, connect args etc. But it's good to start simple.
Anyway, ultimately, I think Airflow really needs to go in this direction: in the new REST API, in the new AIP-72 internal API server, in triggers, and ultimately, in the scheduler.